package com.qs.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * 使用kafka-clients构建生产者，发送数据
 *
 * @author qinyupeng
 * @since 2019-12-19 11:06:22
 */
public class ProducerDemo implements Runnable {

    private final KafkaProducer<Integer, String> producer;

    //topic
    private final String topic;

    //是否异步发送
    private final boolean isAsync;

    public ProducerDemo(String topic, boolean isAsync) {
        Properties properties = new Properties();
        //kafka集群地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.16.109.130:9092,172.16.109.131:9092,172.16.109.132:9092");

        //客户端名称
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "kafkaProducer");

        //设置为"all"，表示消息需要所有的副本确认
        properties.put(ProducerConfig.ACKS_CONFIG, "all");

        //设置一个大于0的值，表示客户端重新发起任何失败消息的次数
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);

        //将多个消息合并发送，设置合并的容量，将减少请求发送的次数，提高效率
        //properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

        //配合batch.size配置，发送消息时，给定延时后发送消息，以便将多个消息一起发送，提高效率
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);

        //key序列化类型
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());

        //value序列化类型
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //构建kafkaProducer对象
        producer = new KafkaProducer<>(properties);

        this.topic = topic;
        this.isAsync = isAsync;
    }

    @Override
    public void run() {
        int num = 0;
        while (num < 50) {
            String message = "message_" + num;
            System.out.println("begin send message:" + message);
            ProducerRecord<Integer, String> record = new ProducerRecord<>(topic, message);
            if (isAsync) {  //异步发送
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (metadata != null) {
                            System.out.println("async-offset:" + metadata.offset() + ", partition:" + metadata.partition());
                        }
                    }
                });
            } else {  //同步发送
                try {
                    Future<RecordMetadata> metadataFuture = producer.send(record);
                    RecordMetadata metadata = metadataFuture.get();
                    System.out.println("async-offset:" + metadata.offset() + ", partition:" + metadata.partition());
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
            }

            num++;

            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
    }

    public static void main(String[] args) {
        ProducerDemo producerDemo = new ProducerDemo("my-replicated-topic", true);
        new Thread(producerDemo).start();
    }
}
